/**
 * Copyright (C) 2016 Eshore Corporation
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package com.eshore.thrpc.server.thrift;

import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocolFactory;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.server.TThreadPoolServer.Args;
import org.apache.thrift.server.TThreadedSelectorServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TServerSocket;
import org.apache.thrift.transport.TTransportException;
import org.apache.thrift.transport.TTransportFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.eshore.thrpc.RpcException;
import com.eshore.thrpc.common.ServerNode;

/**
 * 服务线程
 * <p>
 * 
 * @author : zhouzhixiang
 * @date : 2016-11-13
 */
public class TServerThread extends Thread {

    /** LOGGER */
    private final Logger LOGGER = LoggerFactory.getLogger(getClass());

    /** {@link TServer} */
    private TServer server;

    private boolean nonBlock = true;

    public TServerThread() {}

    /**
     * 使用阻塞方式的服务端
     * 
     * @param processor {@link TProcessor}
     * @param serverNode {@link ServerNode}
     * @param maxWorkerThreads 最大工作线程数
     * @param minWorkerThreads 最小工作线程数
     * @throws RpcException
     */
    protected void createBlockServer(TProcessor processor, ServerNode serverNode,
            int maxWorkerThreads, int minWorkerThreads) throws RpcException {
        nonBlock = false;
        TServerSocket serverTransport;
        try {
            serverTransport = new TServerSocket(serverNode.getPort());
        } catch (TTransportException e) {
            throw new RpcException(RpcException.NETWORK_EXCEPTION, e);
        }
        TProtocolFactory portFactory = new TCompactProtocol.Factory();
        Args args = new Args(serverTransport);
        args.processor(processor);
        args.protocolFactory(portFactory);
        args.maxWorkerThreads(maxWorkerThreads);
        args.minWorkerThreads(minWorkerThreads);
        server = new TThreadPoolServer(args);
        server.setServerEventHandler(new ThriftEventHandler(serverNode));
        setName(serverNode.getName());
    }

    /**
     * @param processor {@link TProcessor}
     * @param serverNode {@link ServerNode}
     * @param maxWorkerThreads 最大工作线程数
     * @param minWorkerThreads 最小工作线程数
     * @param monitor {@link RpcMonitor}
     * @throws RpcException
     */
    protected void createNonBlockThread(TProcessor processor, ServerNode serverNode,
            int selectorThreads, int workerThreads, int acceptQueueSizePerThread)
            throws RpcException {
        this.nonBlock = true;
        TNonblockingServerSocket tnbSocketTransport;
        try {
            tnbSocketTransport =
                    new TNonblockingServerSocket(
                            new TNonblockingServerSocket.NonblockingAbstractServerSocketArgs()
                                    .port(serverNode.getPort()));
        } catch (TTransportException e) {
            throw new RpcException(RpcException.NETWORK_EXCEPTION, e);
        }
        TTransportFactory transportFactory = new TFramedTransport.Factory();
        TProtocolFactory proFactory = new TCompactProtocol.Factory();
        org.apache.thrift.server.TThreadedSelectorServer.Args args =
                new org.apache.thrift.server.TThreadedSelectorServer.Args(tnbSocketTransport);

        LOGGER.info("server workThreads number:{}", workerThreads);

        args.protocolFactory(proFactory);
        args.transportFactory(transportFactory);
        args.processor(processor);
        args.selectorThreads(selectorThreads);
        args.workerThreads(workerThreads);
        args.acceptQueueSizePerThread(acceptQueueSizePerThread);

        server = new TThreadedSelectorServer(args);
        
        server.setServerEventHandler(new ThriftEventHandler(serverNode));
        setName(serverNode.getName());
    }

    /**
     * @param tServer {@link TServer}
     */
    public TServerThread(TServer tServer) {
        this.server = tServer;
    }

    @Override
    public void run() {
        try {
            if (!nonBlock) {
                LOGGER.info("{} is start useing blocking!", getName());
            } else {
                LOGGER.info("{} is start useing nonblocking!", getName());
            }
            server.serve();
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        }
    }

    /**
     * 停止服务
     * <p>
     */
    public void stopServer() {
        server.stop();
        LOGGER.info("{} is stop!", getName());
    }

    /**
     * TServer是否启动
     * <p>
     * 
     * @return
     */
    public boolean isStarted() {
        return server.isServing();
    }

    public void setNonBlock(boolean nonBlock) {
        this.nonBlock = nonBlock;
    }


}
